package cn.flying.cloud.task.core.callback.disruptor;

import javax.annotation.Resource;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.lmax.disruptor.util.DaemonThreadFactory;

import cn.flying.cloud.task.core.callback.JobExecuteCallback;
import cn.flying.cloud.task.core.callback.event.TransferEvent;
import cn.flying.cloud.task.core.callback.factory.TransferEventFactory;
import cn.flying.cloud.task.core.callback.handler.TransferEventHandler;
import cn.flying.cloud.task.core.callback.handler.TransferExceptionHandler;

/**
 * @author: admin
 * @date: 2025年03月18日 15:29
 * @version: 1.0
 */
@Component
public class AsyncJobLogDisruptor implements InitializingBean, DisposableBean {
    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    private Disruptor<TransferEvent> disruptor;
    private TransferEventProducer transferEventProducer;
    private final ThreadPoolExecutor executor = new ThreadPoolExecutor(8, 16, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100));

    @Resource
    private JobExecuteCallback jobExecuteCallback;

    /**
     * @throws Exception
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        logger.info("开始初始化Disruptor！");
        // 1.创建Ring Buffer中事件元素的工厂对象
        TransferEventFactory factory = new TransferEventFactory();
        // 2.指定Ring Buffer的大小,必须为2的幂次方
        int bufferSize = 2048;
        // 3.构造Disruptor
        this.disruptor = new Disruptor<>(factory, bufferSize, DaemonThreadFactory.INSTANCE, ProducerType.MULTI, new BlockingWaitStrategy());
        // 4.设置异常处理
        this.disruptor.setDefaultExceptionHandler(new TransferExceptionHandler());
        // 5.注册消费者
        this.disruptor.handleEventsWith(new TransferEventHandler(jobExecuteCallback));
        // 6.启动Disruptor, 启动线程运行
        this.disruptor.start();
        // 7。初始化ringBuffer
        RingBuffer<TransferEvent> ringBuffer = this.disruptor.getRingBuffer();
        // 8.实例化生产者
        this.transferEventProducer = new TransferEventProducer(ringBuffer);
    }

    /**
     * @throws Exception
     */
    @Override
    public void destroy() throws Exception {
        if (this.disruptor != null) {
            this.disruptor.shutdown();
        }
    }

    /**
     * 发布
     */
    public void publish(String type, String data) {
        this.transferEventProducer.publish(type, data);
    }

    public static class TransferEventProducer {
        private final Logger logger = LoggerFactory.getLogger(this.getClass());

        private final RingBuffer<TransferEvent> ringBuffer;

        public TransferEventProducer(RingBuffer<TransferEvent> ringBuffer) {
            this.ringBuffer = ringBuffer;
        }

        public void publish(String type, String data) {
            // ringBuffer是个队列，其next方法返回的是下最后一条记录之后的位置，这是个可用位置
            long next = ringBuffer.next();
            try {
                TransferEvent event = ringBuffer.get(next);
                event.setType(type);
                event.setData(data);
            } catch (Exception e) {
                logger.error("向RingBuffer队列存入数据出现异常=>", e);
            } finally {
                ringBuffer.publish(next);
            }
        }
    }
}
